

package org.example.flinksql.test;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import java.util.HashMap;
import java.util.Map;


/**
 * 读取配置参数 -> 注册Source、Sink -> 组装任务定义 -> 提交任务
 */
public class StreamJobWithConfig {
	private Map<String, DBConf> configs;

	public StreamJobWithConfig() {
		configs = new HashMap<>();

		HashMap<String, String> tbs = new HashMap<>();
		tbs.put("t_logs", "CREATE TABLE t_logs(" +
				"username STRING," +
				"action STRING," +
				"create_time TIMESTAMP" +
				")");

		DBConf dbConf = new DBConf("flink-test", "under.azure", "3306",
				"root", "123456", tbs);

		configs.put("flink-test", dbConf);
	}

	public String generateCreateTableSql(String db, String tb) {
		String sql = null;

		DBConf dbConf = configs.get(db);
		if (null == dbConf) {
			return sql;
		}

		Map<String, String> tbs = dbConf.getTbs();
		if (null == tbs) {
			return sql;
		}

		String tbConf = tbs.get(tb);
		String dbName = dbConf.getDbName();
		String host = dbConf.getHost();
		String port = dbConf.getPort();
		String username = dbConf.getUsername();
		String password = dbConf.getPassword();

		sql = String.format("%s WITH (" +
				"'connector' = 'jdbc'," +
				"'url' = 'jdbc:mysql://%s:%s/%s?useSSL=false'," +
				"'table-name' = '%s'," +
				"'username' = '%s'," +
				"'password' = '%s'" +
				")", tbConf, host, port, dbName, tb, username, password);

		return sql;
	}

	public class DBConf {
		private String dbName;
		private String host;
		private String port;
		private String username;
		private String password;
		private Map<String,String> tbs;

		public DBConf(String dbName, String host, String port, String username, String password, Map<String,String> tbs) {
			this.dbName = dbName;
			this.host = host;
			this.port = port;
			this.tbs = tbs;
			this.username = username;
			this.password = password;
		}

		public String getDbName() {
			return dbName;
		}

		public String getHost() {
			return host;
		}

		public String getPort() {
			return port;
		}

		public String getUsername() {
			return username;
		}

		public String getPassword() {
			return password;
		}

		public Map<String, String> getTbs() {
			return tbs;
		}
	}


	public static void main(String[] args) throws Exception {
		//args:["flink-test","t_logs","where username = 'robbin'"]
		if (null == args || args.length == 0) {
			System.err.println("no args!");
			return;
		}

		EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
		TableEnvironment tbEnv = TableEnvironment.create(bsSettings);


		String dbName = args[0];
		String tbName = args[1];
		String filter = args[2];

		System.out.println(String.format("db:%s tb:%s filter:%s", dbName, tbName, filter));

		StreamJobWithConfig job = new StreamJobWithConfig();
		String createSql = job.generateCreateTableSql(dbName, tbName);

		tbEnv.executeSql(createSql);

		String querySql = String.format("select * from %s %s", tbName, filter.startsWith("where") ? filter : "");

		Table logs = tbEnv.sqlQuery(querySql);

		logs.execute().print();

		// execute program
//		env.execute("Flink Streaming Java API Skeleton");
	}
}
